package com.atguigu.flink.streamapi.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * Created by Smexy on 2022/11/21
 *
 *  官方不提供。
 *          选择第三方社区提供的。
 *
 *   String,Set,List: 是一种套路
 *
 *   Hash,Zset: 是另外一种套路
 *
 *
 *   RedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RedisMapper<IN> redisSinkMapper)
 *          FlinkJedisConfigBase: 配置如何连接redis
 *          RedisMapper：   配置如何向redis写入。  类型，key，value
 */
public class Demo4_RedisSink
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //FlinkJedisPoolConfig
        FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop102")
            .setPort(6379)
            .setDatabase(0)
            .setMaxTotal(20)
            .setMinIdle(5)
            .setMaxIdle(10)  // 池中没有人来借连接时，依旧准备 5-10 个空闲连接
            .setTimeout(60000)
            .build();


        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .addSink(new RedisSink<>(flinkJedisPoolConfig,new MyZsetMapper()));



        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    /*
        写String

        RedisCommandDescription getCommandDescription()： 用什么命令写入redis。命令决定了value的类型
                set: string
                sadd: set
                rpush,lpush: list

                public RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： Hash,Zset: 是另外一种套路
                public RedisCommandDescription(RedisCommand redisCommand)：   String,Set,List: 是一种套路


            String getKeyFromData: 从数据中抽取一个字段，作为key
            String  getValueFromData: 从数据中抽取一个字段，作为value
     */
    private static class MyZsetMapper implements RedisMapper<WaterSensor>{

        /*
                zset ：  key
                        score member

                   以vc作为分数
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            //所有的传感器保存一个key-value
            return new RedisCommandDescription(RedisCommand.ZADD,"myZset");
        }

        //member
        @Override
        public String getKeyFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }

        //score
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.getVc().toString();
        }
    }

    private static class MyHashMapper implements RedisMapper<WaterSensor>{

        /*
                hash ：  key
                        value  {
                                    field  value
                               }

                   所有的传感器都保存到一个K-V，每种id的传感器只保存一个，最新的
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            //所有的传感器保存一个key-value
            return new RedisCommandDescription(RedisCommand.HSET,"myHash");
        }

        //field
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //value
        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }

    private static class MySetMapper implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SADD);
        }

        @Override
        public String getKeyFromData(WaterSensor data) {
            return "wsset";
        }

        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }

    private static class MyListMapper implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        @Override
        public String getKeyFromData(WaterSensor data) {
            return "wslist";
        }

        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }

    private static class MyStringMapper implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SET);
        }

        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }
}
